package com.mindata.blockchain.socket.raft.server;

import com.mindata.blockchain.socket.raft.core.proto.RaftMessage;
import com.github.wenweihu86.rpc.client.RPCClient;
import com.github.wenweihu86.rpc.client.RPCProxy;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import com.mindata.blockchain.socket.raft.core.RaftNode;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author:wenhui
 * @Description:
 * @Date:Create in 2018-7-25 16:12
 **/
@Slf4j
public class RaftServerServiceImpl implements RaftServerService {
    private static final Logger LOG = LoggerFactory.getLogger(RaftServerServiceImpl.class);
    private static JsonFormat.Printer printer = JsonFormat.printer().omittingInsignificantWhitespace();

    private RaftNode raftNode;
    private RaftStateMachine stateMachine;

    public RaftServerServiceImpl(RaftNode raftNode, RaftStateMachine stateMachine) {
        this.raftNode = raftNode;
        this.stateMachine = stateMachine;
    }

    @Override
    public ServerMessage.SetResponse set(ServerMessage.SetRequest request) {
        ServerMessage.SetResponse.Builder responseBuilder = ServerMessage.SetResponse.newBuilder();
        // 如果自己不是leader，将写请求转发给leader
        if (raftNode.getLeaderId() <= 0) {
            responseBuilder.setSuccess(false);
            log.warn("raft leader 不存在");
        } else if (raftNode.getLeaderId() != raftNode.getLocalServer().getServerId()) {
            RPCClient rpcClient = raftNode.getPeerMap().get(raftNode.getLeaderId()).getRpcClient();
            RaftServerService serverService = RPCProxy.getProxy(rpcClient, RaftServerService.class);
            ServerMessage.SetResponse responseFromLeader = serverService.set(request);
            responseBuilder.mergeFrom(responseFromLeader);
        } else {
            // 数据同步写入raft集群
            byte[] data = request.toByteArray();
            boolean success = raftNode.replicate(data, RaftMessage.EntryType.ENTRY_TYPE_DATA);
            responseBuilder.setSuccess(success);
        }

        ServerMessage.SetResponse response = responseBuilder.build();
        try {
            LOG.info("set request, request={}, response={}", printer.print(request),
                    printer.print(response));
        } catch (InvalidProtocolBufferException ex) {
            LOG.error("set error", ex);
        }
        return response;
    }

    @Override
    public ServerMessage.GetResponse get(ServerMessage.GetRequest request) {
        ServerMessage.GetResponse response = stateMachine.get(request);
        try {
            LOG.info("get request, request={}, response={}", printer.print(request),
                    printer.print(response));
        } catch (InvalidProtocolBufferException ex) {
            LOG.error("get error", ex);
        }
        return response;
    }
}
